package cn.zwx.common.canal.config;

import cn.zwx.base.exception.BizWrapException;
import cn.zwx.common.canal.constants.CanalConstants;
import cn.zwx.common.canal.enums.CanalCodeEnum;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.List;

/**
 * @author zhangwenxe
 */
@Configuration
@EnableConfigurationProperties(CanalPropertiesConfig.class)
public class CanalClientConfig {

    public static final Logger logger = LoggerFactory.getLogger(CanalClientConfig.class);

    @Resource
    public CanalPropertiesConfig propertiesConfig;


    @PostConstruct
    public void builderCanalConnector() throws BizWrapException {
       final List<CanalPropertiesConfig.CanalInstance> instanceList = propertiesConfig.getInstances();
       if (CollectionUtils.isEmpty(instanceList)){
           final CanalCodeEnum initCanalError = CanalCodeEnum.INIT_CANAL_ERROR;
           final String initCanalDesc = String.format(initCanalError.getDesc(), "没有配置instances信息,请检查配置");
           throw new BizWrapException(initCanalError.getCode(),initCanalDesc);
       }
        for (CanalPropertiesConfig.CanalInstance canalInstance : instanceList) {
            final CanalConnector canalConnector = this.initCanalConnector(canalInstance.getDestination(), canalInstance.getSubscribe());
            CanalConstants.canalConnectorMap.putIfAbsent(canalInstance.getDestination(),canalConnector);
        }
        logger.info("builderCanalConnector success !!");
    }

    public CanalConnector initCanalConnector(String destination, String subscribe) throws BizWrapException {
        logger.info("init destination:{},Subscribe:{} start !!!",destination,subscribe);
        final boolean clusterEnabled = propertiesConfig.isClusterEnabled();
        CanalConnector canalConnector;
        if (clusterEnabled){
            canalConnector = CanalConnectors.newClusterConnector(propertiesConfig.getServerAddr() ,destination,propertiesConfig.getUsername(), propertiesConfig.getPassword() );
        }else {
            String[] arr = propertiesConfig.getServerAddr().split(":");
            InetSocketAddress addr =  new InetSocketAddress(arr[0], Integer.parseInt(arr[1]));
            canalConnector = CanalConnectors.newSingleConnector(addr,destination,propertiesConfig.getUsername(), propertiesConfig.getPassword());
        }
        try {
            // 2.创建连接
            canalConnector.connect();
            // 3.订阅客户端 客户端指定会覆盖服务指定配置
            canalConnector.subscribe(subscribe);
            // 4.回滚寻找上次中断的位置
            canalConnector.rollback();
        } catch (Throwable t) {
            logger.error("failed to connect to canal server:{}", destination);
            canalConnector.disconnect();
            final CanalCodeEnum initCanalError = CanalCodeEnum.INIT_CANAL_ERROR;
            final String initCanalDesc = String.format(initCanalError.getDesc(), "链接实例:"+destination+",请检查配置");
            throw new BizWrapException(initCanalError.getCode(),initCanalDesc);
        }
        logger.info("init destination:{},subscribe:{} success !!!",destination,subscribe);
        return canalConnector;
    }

}
